<?php

class Worker {
	
	protected  $task;
	protected  $taskInfo;
	protected  $lockFile;
	protected  $taskName;

	protected  $execution;	
	protected  $taskService;
	protected  $schedulerService;
	
	public function __construct($basicInfo) {
		$this->taskInfo = $basicInfo;

		$this->taskName = "task_{$this->taskInfo['jobId']}_{$this->taskInfo['taskId']}";
		
		$this->taskService = TaskService::getInstance();
		$this->schedulerService = SchedulerService::getInstance();	
	}
	
	public function run() {
        Yii::log('begin run worker for task: '. $this->taskName, CLogger::LEVEL_INFO, WORKER_LOG);

        if(($ret = $this->getLock($this->taskInfo)) !== true) {
            Yii::log('unable to get worker lock of task: '.$this->taskName . var_export($this->taskInfo, true), CLogger::LEVEL_ERROR, WORKER_LOG);
            return false;
        }
        Yii::log('get worker lock success for task: '.$this->taskName, CLogger::LEVEL_INFO, WORKER_LOG);   

        $ret = $this->updateTaskState('task is preparing');
        if(($ret = $this->prepare()) !== true) {
            Yii::log("prepare worker failed for task: ".$this->taskName, CLogger::LEVEL_ERROR, WORKER_LOG);
            $this->safeEndTask(TaskState::Failed, "task prepare failed", array());
            return false;
        }
        
        $ret = $this->updateTaskState('task is executing');
        $ret = $this->execute();
        
        if(!isset($this->execution)) {
        	$this->safeEndTask(TaskState::Failed, 'excpetion occur when execute workflow', array());
        	return false;
        }
        
        $runtimeInfo = $this->execution->getVariable('runtimeInfo');
        if($ret === false) {
        	$this->safeEndTask($runtimeInfo['state'], $runtimeInfo['message'], $runtimeInfo);
        	return false;
        }
        $this->safeEndTask(TaskState::Done, 'task executed successfully', $runtimeInfo);
        exec("rm -rf " . $this->taskInfo['rtPath']);
        Yii::log('task executed successfully: '.$this->taskName, CLogger::LEVEL_INFO, WORKER_LOG);
        return true;
	}
	
	protected function prepare() {
    	$this->taskInfo['pathDate'] = DateHelper::makePathDate(strtotime($this->taskInfo['baseDate']));
        $this->taskInfo['rtPath'] = PathUtil::getRuntimePath(
            $this->taskInfo['taskId'], 
            $this->taskInfo['jobId'], 
            $this->taskInfo['pathDate']
        );
        
        $this->task = $this->taskService->getTaskById($this->taskInfo['taskId']);
	 	if(!isset($this->task)) {
            Yii::log("can't find task[{$this->taskInfo['taskId']}] record in db", CLogger::LEVEL_ERROR, WORKER_LOG);
            return false;
        }
 
		// make dir
		if(!file_exists("{$this->taskInfo['rtPath']}/log")){
			mkdir("{$this->taskInfo['rtPath']}/log");
		}
		if(!file_exists("{$this->taskInfo['rtPath']}/output")) {
			mkdir("{$this->taskInfo['rtPath']}/output");
		}
		if(!file_exists("{$this->taskInfo['rtPath']}/file")){
			mkdir("{$this->taskInfo['rtPath']}/file");
		}
		if(!file_exists("{$this->taskInfo['rtPath']}/conf")) {
			mkdir("{$this->taskInfo['rtPath']}/conf");
		}
		if(!file_exists("{$this->taskInfo['rtPath']}/variable")) {
			mkdir("{$this->taskInfo['rtPath']}/variable");
		}
		
		// Prepare some script file.
		if(!file_exists("{$this->taskInfo['rtPath']}/bin")) {
			symlink(dirname(__FILE__) . '/../script', "{$this->taskInfo['rtPath']}/bin");
		} 

		putenv("JOBID={$this->taskInfo['jobId']}");
		putenv("TASKID={$this->taskInfo['taskId']}");
		putenv("WORKFLOWNAME={$this->taskInfo['workflowName']}");
		putenv("RUNTIME={$this->taskInfo['rtPath']}");
		putenv("BASEDATE={$this->taskInfo['baseDate']}");
		putenv("FREQUENCY={$this->taskInfo['frequency']}");
		
        return true;
    }
    
    protected function execute() {
    	Yii::log("begin execute task[{$this->taskName}]", CLogger::LEVEL_INFO, WORKER_LOG);
    	
		try {
			// Set up database connection.
			$connectionString = StringUtil::getConnectionString(Yii::app()->db->connectionString, 
				Yii::app()->db->driverName, 
				Yii::app()->db->username, 
				Yii::app()->db->password);
			$db = ezcDbFactory::create($connectionString);

			// Set up workflow definition storage (database).
			$definition = new ezcWorkflowDatabaseDefinitionStorage($db);

			// Load latest version of workflow by id and set variable handler.
			$workflow = $definition->loadByName($this->taskInfo['workflowName']);
			//$workflow->addVariableHandler('params', 'VariableHandler');

			if(isset($this->task['execution_id']) && !empty($this->task['execution_id']) && $this->task['state'] != TaskState::Done) {
				// Resume workflow execution
				Yii::log("resume task[{$this->taskName}] workflow execution[{$this->task['execution_id']}]", CLogger::LEVEL_INFO, WORKER_LOG);
				$this->execution = new ezcWorkflowDatabaseExecution($db, intval($this->task['execution_id']));
				$this->execution->resume();
			}
			else {
				// Set up database-based workflow executer.
				$this->execution = new ezcWorkflowDatabaseExecution($db);
				$this->execution->workflow = $workflow;
				$this->execution->setVariable('runtimeInfo', array(
					'message'	=>	'',
					'state'		=>	TaskState::Running,
					'cost'		=>	array(),
				));
				
				// Start workflow execution.
				$executionId = $this->execution->start();
				Yii::log("start task[{$this->taskName}] workflow execution[$executionId]", CLogger::LEVEL_INFO, WORKER_LOG);
				$this->taskService->editTaskAttributes($this->taskInfo['taskId'], array('execution_id'   =>  $executionId));
			}
			// Save task result info
			/*
			$this->resultInfo[$key] = $receiver->getTaskResult();
			$this->taskService->editTaskAttributes($this->taskInfo['taskId'], array(
				'result_info' =>  CJSON::encode($this->resultInfo[$key])
				)
			);
			*/
			if(!$this->execution->hasEnded()) {
				Yii::log('workflow execution is not ended for task:'.$this->taskName, CLogger::LEVEL_ERROR, WORKER_LOG);
				return false;
			}
		}
		catch(Exception $e) {
			Yii::log('execute task error:' . $e->getMessage(), CLogger::LEVEL_ERROR, WORKER_LOG);
			return false;
		}
    	Yii::log('execute task successfully:'.$this->taskName, CLogger::LEVEL_INFO, WORKER_LOG);
        return true;
    }
    
 	private function getLock() {
        $lock_file = NULL;

        $path_date = DateHelper::makePathDate(strtotime($this->taskInfo['baseDate']));
        $lock_path = LOCK_PATH."/worker_{$path_date}_{$this->taskInfo['taskId']}.pid";

        if (file_exists($lock_path)) {
            $last_pid = intval(file_get_contents($lock_path));
        } 
        else {
            touch($lock_path);
        }

        if(!($lock_file = fopen($lock_path, 'r+'))) {
            Yii::log("unable to open $lock_path", CLogger::LEVEL_ERROR, WORKER_LOG);
            return array('last_pid' => $last_pid, 'detail' => 'open lock_file error');
        }

        if(!flock($lock_file, LOCK_EX|LOCK_NB)) {
            Yii::log("unable to obtain lock, last_pid=$last_pid, detail:lock file error", CLogger::LEVEL_ERROR, WORKER_LOG);

            return array('last_pid' => $last_pid, 'detail' => 'lock lock_file error');
        } 
        else {
            $this->lockFile = $lock_file;
        }
        file_put_contents($lock_path, posix_getpid());
        return true;
    }

    private function releaseLock() {
        $lock_path = LOCK_PATH."/worker_{$this->taskInfo['pathDate']}_{$this->taskInfo['taskId']}.pid";
        if ($this->lockFile) {
            fclose($this->lockFile);
        }
        unlink($lock_path);
    }
    
 	private function updateTaskState($taskDetail) {
		if(is_array($taskDetail)) {
			$taskDetail = CJSON::encode($taskDetail);
		}
        Yii::log("update task state: $taskDetail, ". $this->taskName,  CLogger::LEVEL_INFO, WORKER_LOG);

        return $this->taskService->editTaskAttributes($this->taskInfo['taskId'], array('detail'   =>  $taskDetail));
    }
    
	private function safeEndTask($state, $detail, $runtimeInfo) {
        $row = array(
            'state' => $state,
            'detail' => $detail,
        	'runtime_info'	=>	CJSON::encode($runtimeInfo),
			'end_time'	=> date('Y-m-d H:i:s'),
        );
        
        while($this->taskService->editTaskAttributes($this->taskInfo['taskId'], $row) === false) {
  			Yii::log("failed to end task safely, keep retrying... " . $this->taskName,
				CLogger::LEVEL_ERROR, WORKER_LOG);          
			sleep(2);
        }
        
		while($this->schedulerService->updateCurrentCount($this->taskInfo['executorId'], false) === false) {
  			Yii::log("failed to end task safely, keep retrying... " . $this->taskName,
				CLogger::LEVEL_ERROR, WORKER_LOG);          
			sleep(2);
        }
        
        $this->releaseLock();
    }
}
